package com.bsj.travel.mqtt.listener;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.net.NetUtil;
import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.bsj.travel.cached.redis.RedisCached;
import com.bsj.travel.config.rocketmq.producer.BaseMqPushUtil;
import com.bsj.travel.constant.ActConstants;
import com.bsj.travel.constant.BaseServerTopicConstants;
import com.bsj.travel.constants.RedisConstant;
import com.bsj.travel.def.common.DTO.mqtt.EventAct;
import com.bsj.travel.def.common.DTO.mqtt.MqttEvent;
import com.bsj.travel.def.common.DTO.mqtt.NetworkEnum;
import com.bsj.travel.def.mqtt.message.MqttConnectOfflineMsg;
import com.bsj.travel.mqtt.enums.RedisKeys;
import com.bsj.travel.util.ip.GetEth0IpUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import net.dreamlu.iot.mqtt.core.server.MqttServer;
import net.dreamlu.iot.mqtt.core.server.MqttServerCreator;
import net.dreamlu.iot.mqtt.core.server.event.IMqttConnectStatusListener;
import net.dreamlu.mica.core.utils.INetUtil;
import net.dreamlu.mica.redis.cache.MicaRedisCache;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.tio.core.ChannelContext;

import java.util.*;

/**
 * @author yinhao
 * @version 1.0
 * @description 监听Mqtt连接状态
 * @date 2023/11/7
 */
@Slf4j
@RequiredArgsConstructor
public class MqttConnectStatusListener implements IMqttConnectStatusListener, SmartInitializingSingleton, DisposableBean {
    private final ApplicationContext context;
    private final MicaRedisCache micaRedisCache;
    private final RedisCached redisCached;
    /**
     * mqtt 服务端参数构造
     */
    private MqttServerCreator serverCreator;
    /**
     * mqtt 服务端
     */
    private MqttServer mqttServer;
    /**
     * RocketMq消息发送
     */
    private final BaseMqPushUtil baseMqPushUtil;

    /**
     * 客户端上线
     *
     * @param channelContext
     * @param clientId
     * @param username
     */
    @Override
    public void online(ChannelContext channelContext, String clientId, String username) {
        log.info("MqttClientOnline clientId: {}  username: {}", clientId, username);
        // 将在线的设备踢下线，给新的连接上线
        doMqttConnectOffline(clientId, username);
        // 上线时间
        long millis = System.currentTimeMillis();
        MqttEvent<String> mqttEvent = new MqttEvent<>();
        mqttEvent.setClient(username);
        // 客户端上线IP
        String ip = channelContext.getClientNode().getIp();
        // 把上线IP放到data里面
        mqttEvent.setData(ip);
        EventAct eventAct = EventAct.builder()
                .timestamp(millis)
                .traceId(RandomUtil.randomString(30))
                .serveType(2)
                .operate(ActConstants.ONLINE)
                .code(NetworkEnum.SUCCESS.getCode())
                .event(mqttEvent)
                .build();
        baseMqPushUtil.sendTopicBytes(BaseServerTopicConstants.TRAVEL_MQTT_MSG_PROCESS_TOPIC,
                BaseServerTopicConstants.TRAVEL_MQTT_MSG_PROCESS_BEHAVIOR_TAG, JSON.toJSONBytes(eventAct));
        // 连接状态存储
        micaRedisCache.hSet(getRedisKey(), username, clientId);
        // 设备所在节点存储
        redisCached.set(RedisKeys.CONNECT_NODE.getKey(username), serverCreator.getIp());
    }

    /**
     * 客户端离线
     *
     * @param channelContext
     * @param clientId
     * @param username
     * @param reason
     */
    @Override
    public void offline(ChannelContext channelContext, String clientId, String username, String reason) {
        log.info("MqttClientOffline clientId: {}  username: {}  reason: {},closeCode:{}", clientId, username, reason, channelContext.getCloseCode());
        long millis = System.currentTimeMillis();
        MqttEvent mqttEvent = new MqttEvent();
        mqttEvent.setClient(username);
        EventAct eventAct = EventAct.builder()
                .timestamp(millis)
                .traceId(RandomUtil.randomString(30))
                .serveType(2)
                .operate(ActConstants.OFFLINE)
                .code(NetworkEnum.SUCCESS.getCode())
                .event(mqttEvent)
                .build();
        baseMqPushUtil.sendTopicBytes(BaseServerTopicConstants.TRAVEL_MQTT_MSG_PROCESS_TOPIC,
                BaseServerTopicConstants.TRAVEL_MQTT_MSG_PROCESS_BEHAVIOR_TAG, JSON.toJSONBytes(eventAct));
        String clientStr = micaRedisCache.hGet(getRedisKey(), username);
        // 判断这个clientId是不是自己，不是的话，就不管
        if (clientId.equalsIgnoreCase(clientStr)) {
            micaRedisCache.hDel(getRedisKey(), username);
        }
        // 删除设备所在节点存储
        redisCached.del(RedisKeys.CONNECT_NODE.getKey(username));
    }

    /**
     * 该节点服务端停机
     */
    @Override
    public void destroy() {
        // 删除当前节点下，设备所在节点存储
        Map<String, String> all = redisCached.hgetAll(getRedisKey());
        if (CollectionUtil.isNotEmpty(all)) {
            List<String> keys = new ArrayList<>();
            for (Map.Entry<String, String> allNext : all.entrySet()) {
                String username = allNext.getKey();
                keys.add(RedisKeys.CONNECT_NODE.getKey(username));
            }
            List<List<String>> batDelKeys = CollectionUtil.split(keys, 500);
            batDelKeys.forEach(v -> {
                if (v != null) {
                    micaRedisCache.del(v);
                }
            });
        }

        // 停机删除集合
        micaRedisCache.del(getRedisKey());
    }

    /**
     * 获取该节点服务端信息
     */
    @Override
    public void afterSingletonsInstantiated() {
        this.serverCreator = context.getBean(MqttServerCreator.class);
        // 设置服务器IP，避免每次获取节点IP都要重新获取
        this.serverCreator.ip(GetEth0IpUtil.getHostAddress());
        this.mqttServer = context.getBean(MqttServer.class);
    }

    /**
     * 获取节点
     *
     * @return
     */
    private String getRedisKey() {
        return RedisKeys.CONNECT_STATUS.getKey(serverCreator.getIp());
    }

    /**
     * 将在线的设备踢下线，给新的连接上线
     *
     * @param clientId 客户端ID
     * @param username 用户名
     */
    public void doMqttConnectOffline(String clientId, String username) {
        String nodeName = redisCached.get(RedisKeys.CONNECT_NODE.getKey(username));
        if (nodeName != null) {
            // 如果设备在线，则判断是否在当前节点，存在则关闭
            if (nodeName.equals(serverCreator.getIp())) {
                mqttServer.close(clientId);
            } else {
                // 如果设备在其他节点，则推送消息通知其他节点进行关闭
                String channel = RedisConstant.LINE_PUB_SUB_TOPIC + nodeName;
                log.info("旧链接强踢下线channel: {}, clientId: {}, username:{}", channel, clientId, username);
                MqttConnectOfflineMsg mqttConnectOfflineMsg = new MqttConnectOfflineMsg();
                mqttConnectOfflineMsg.setUsername(username);
                mqttConnectOfflineMsg.setClientId(clientId);
                redisCached.pub(channel.getBytes(), JSONObject.toJSONString(mqttConnectOfflineMsg).getBytes());
            }
        }
    }
}
